/**
 * 消费debezium 消息
 * https://kafka.js.org/docs/consuming
 * @author xuejianxin
 * @date 2022-12-02
 */

const { kafka } = require('../common/kafka')

async function run () {
// 新建一个消费者,一点要注意groupId 不能和别的应用重复
  const consumer = kafka.consumer({
    groupId: 'my-group'
    // sessionTimeout: 30000, //默认 30000，用于检测故障的超时时间（以毫秒为单位）。消费者定期发送心跳以向代理指示其活跃度。如果在此会话超时到期之前代理没有收到心跳，则代理将从组中删除此消费者并启动重新平衡
    // heartbeatInterval: 3000, // 默认 3000, 消费者协调器的心跳之间的预期时间（以毫秒为单位）。心跳用于确保消费者的会话保持活动状态。该值必须设置为低于会话超时
    // autoCommit: true,//默认：true 自动提交。相反，可以手动提交 offsets。
    // autoCommitInterval：5000 ,//消费者将在给定时间段（例如五秒）后提交偏移量。以毫秒为单位的值。
    // autoCommitThreshold：// 消费者会在解析给定数量的消息后提交偏移量，例如一百条消息。默认：null
  })
  // 等待连接
  await consumer.connect()

  // await consumer.subscribe({ topics: ['bj.test.demo', 'bj.test.user_tel'], fromBeginning: true })
  // await consumer.subscribe({ topics: [/topic-(eu|us)-.*/i] })
  // 使用正则表达式时，消费者不会匹配订阅后创建的主题。
  // 如果使用订阅/topic-.*/时 broker只存在 topic-A，topic-B ，
  // 然后topic-C被创建，消费者将不会自动订阅topic-C。
  await consumer.subscribe({ topics: [/bj.test.*/i], fromBeginning: true })

  await consumer.run({
    // eachMessage一次处理一条消息。它是在eachBatch之上实现的，它会在配置的时间间隔内自动为您提交偏移量和心跳
    // eachMessage处理程序的阻塞时间不应超过配置的会话超时时间sessionTimeout，否则消费者将从组中删除。
    // 如果您的工作负载涉及单个消息的处理时间非常慢，那么应该增加会话超时或定期使用heartbeat处理程序有效负载中公开的函数。
    eachMessage: async ({ topic, partition, message, heartbeat, pause }) => {
    // 处理业务
      console.log({
        key: message.key?.toString(),
        value: message.value?.toString(),
        headers: message.headers
      })
    }
  })
}
// 开始运行 npm run kafka
run()
